// Copyright (c) 2023 Huawei Device Co., Ltd.
// Licensed under the Apache License, Version 2.0 (the "License");
// you may not use this file except in compliance with the License.
// You may obtain a copy of the License at
//
//     http://www.apache.org/licenses/LICENSE-2.0
//
// Unless required by applicable law or agreed to in writing, software
// distributed under the License is distributed on an "AS IS" BASIS,
// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
// See the License for the specific language governing permissions and
// limitations under the License.

use crate::error::Error;
use std::future::Future;
use std::sync::Arc;
use tokio::runtime::{Builder, Handle, Runtime};
use tokio::sync::Notify;
use tokio::task::JoinHandle;

/// Dispatcher schedules the actors running on it.
pub struct Dispatcher {
    executor: Runtime,
    stop: Arc<Notify>,
}

impl Dispatcher {
    /// Creates a dispatcher with single thread. User should call `run()` to run the dispatcher.
    /// See [`Dispatcher::run`].
    pub fn new() -> Result<Self, Error> {
        Ok(Self {
            executor: Builder::new_current_thread()
                .enable_all()
                .build()
                .map_err(|e| Error::CreateRuntime(e.to_string()))?,
            stop: Arc::new(Notify::new()),
        })
    }

    /// Creates a dispatcher with single thread. A new thread will be created and the dispatcher
    /// will run automatically on it.
    pub fn new_detached(thread_name: &str) -> Result<DispatcherHandle, Error> {
        let dispatcher = Dispatcher::new()?;
        let handle = dispatcher.handle();
        std::thread::Builder::new()
            .name(thread_name.to_string())
            .spawn(move || dispatcher.run())
            .map_err(|e| Error::CreateThread(e.to_string()))?;
        Ok(handle)
    }

    /// Returns a handle to the dispatcher. The returned handle can be used to start actors on this
    /// dispatcher, to stop the dispatcher, and can be cloned to allow moving the `DispatcherHandle`
    /// to other threads.
    /// User should call this method before calling [`Dispatcher::run`].
    pub fn handle(&self) -> DispatcherHandle {
        DispatcherHandle {
            handle: self.executor.handle().clone(),
            stop: self.stop.clone(),
        }
    }

    /// Spawns a task to the dispatcher.
    pub(crate) fn spawn<F>(&self, future: F) -> JoinHandle<F::Output>
    where
        F: Future + Send + 'static,
        F::Output: Send + 'static,
    {
        self.executor.spawn(future)
    }

    /// Runs the dispatcher, all the actors on it will start to work too. This function blocks the
    /// current thread, and will never return until the [`DispatcherHandle::stop`] function is
    /// called.
    /// User should call [`Dispatcher::handle`] before calling this method.
    ///
    /// # Panics
    ///
    /// This function panics if called within an asynchronous execution context.
    pub fn run(self) {
        self.executor.block_on(async {
            self.stop.notified().await;
        });
    }
}

/// Handle to the dispatcher. The Handle can be freely cloned.
#[derive(Clone)]
pub struct DispatcherHandle {
    handle: Handle,
    stop: Arc<Notify>,
}

impl DispatcherHandle {
    /// Stops the dispatcher, and all the actors running on it will be dropped, all the msgs haven't
    /// been processed will be dropped too.
    pub fn stop(self) {
        self.stop.notify_one();
    }

    /// Spawns a task to the dispatcher.
    pub(crate) fn spawn<F>(&self, future: F) -> JoinHandle<F::Output>
    where
        F: Future + Send + 'static,
        F::Output: Send + 'static,
    {
        self.handle.spawn(future)
    }
}
